package com.elinshaw.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 行键过滤器
 */
public class RowKeyFilter {

    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "datanode1:2181,datanode1:2181,datanode1:2181");
        Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(TableName.valueOf("t2"));
        Scan scan = new Scan();
        //筛选出行键为row2的一行数据
        //Filter filter = new RowFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("row2")));
        //正则比较器以row开头的字符串
        //Filter filter = new RowFilter(CompareOperator.EQUAL, new RegexStringComparator("row."));
        //列族过滤器
        // Filter filter = new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("f1")));
        //列过滤器：筛选出列为name的所有数据
        //Filter filter = new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("name")));
        //值过滤器：筛选出一行中的值包含"beijing"的所有单元格数据
        //Filter filter = new ValueFilter(CompareOperator.EQUAL, new SubstringComparator("beijing"));
        //单列值过滤器：用一列的值决定该行是否被过滤
        //筛选出name列不包含xiaoming的所有行数据
        Filter filter = new SingleColumnValueFilter(Bytes.toBytes("f1"), Bytes.toBytes("name"), CompareOperator.NOT_EQUAL, new SubstringComparator("xiaoming"));
        //如果某行列name不存在，那么该行将被过滤掉，false则不进行过滤，默认为false。
       // ((SingleColumnValueFilter) filter).setFilterIfMissing(true);
        scan.setFilter(filter);
        //查询数据
        ResultScanner rs = table.getScanner(scan);
        for (Result res : rs) {
            System.out.println(res);
        }
        rs.close();

    }


}
